package com.inspinia.base.mq.rabbitmq.service;

import com.inspinia.base.mq.rabbitmq.config.RabbitConfig;
import com.inspinia.base.mq.rabbitmq.config.RabbitMqConfig;
import com.inspinia.base.util.JsonUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Component;
import redis.clients.util.SafeEncoder;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;

import static org.springframework.amqp.rabbit.connection.RabbitUtils.closeChannel;

/**
 * @author Veblen
 */
@Component
@EnableAsync
@Slf4j
public class RabbitMqMessageSender implements RabbitTemplate.ConfirmCallback {

    private static Connection connection = RabbitMqConfig.getConnection();

    /**
     * 异步发送消息到队列
     * @param queueName 队列名称
     * @param message 消息内容
     */
    public static void sendMsg(String queueName,String message) {
        Channel channel = null;
        try {
            channel = connection.createChannel();
            channel.queueDeclare(queueName, true, false, false, null);
            channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, SafeEncoder.encode(message));
        } catch (Exception e) {
            log.error("消息发送失败", e);
        } finally {
            closeChannel(channel);
        }
    }

    /**
     * 发送消息方法，可以直接传Bean
     *
     * @param queueName
     * @param data
     * @param <T>
     * @return
     */
    public static <T> void sendMsg(String queueName, T data) {
        String message = "";
        //如果是基本类型，直接处理
        if (data.getClass().isPrimitive()) {
            message = data.toString();
            sendMsg(queueName, message);
        } else {
            message = JsonUtils.toJson(data);
            sendMsg(queueName, message);
        }
    }

    /**
     * 发送延迟消息到队列
     * @param queueName 队列名称
     * @param message 消息
     * @param delayTimeInsSecond 延迟多少秒
     * @return
     */
    public static void sendDelayMessage(String queueName,String message,long delayTimeInsSecond){
        Channel channel = null;
        try {
            channel = connection.createChannel();
            byte[] messageBodyBytes = message.getBytes("UTF-8");
            Map<String,Object> headers = new ConcurrentSkipListMap<>();
            headers.put("x-delay",1000 * delayTimeInsSecond);
            AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder();
            channel.queueDeclare(queueName,true,false,false,null);
            channel.basicPublish(RabbitConfig.DELAY_QUEUE_EXCHANGE, queueName, props.build(), messageBodyBytes);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送延迟消息方法，可以直接传Bean
     *
     * @param routingKey 路由的key
     * @param data 对象
     * @param delayTimeInSecond 延迟事件
     * @param <T> 对象类型
     */
    public static  <T> void sendDelayMessage(String routingKey,  T data, long delayTimeInSecond) {
        log.info("TraceId_Elisa_rabbitMQ queueName {} uid {} delayTimeInSecond {}",routingKey, JsonUtils.toJson(data),delayTimeInSecond);
        String objectJson = "";
        //如果是基本类型，直接处理
        if (data.getClass().isPrimitive()) {
            objectJson = data.toString();
            sendDelayMessage(routingKey,objectJson, delayTimeInSecond);
        } else {
            objectJson = JsonUtils.toJson(data);
            sendDelayMessage(routingKey,objectJson, delayTimeInSecond);
        }
    }


    /**
     * 消息消费确认回调
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("回调id:" + correlationData);
        if (ack) {
            System.out.println("消息成功消费");
        } else {
            System.out.println("消息消费失败:" + cause);
        }
    }
}